package com.bw.yk01;

import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Properties;

import static org.apache.flink.table.api.Expressions.$;

/*
(1,192.168.0.1,fail,1558430842) A
(1,192.168.0.2,fail,1558430843) B
1,192.168.0.2,fail,1558430844
1,192.168.0.3,fail,1558430850
1,192.168.0.3,fail,1558430851
2,192.168.10.10,fail,1558430851
2,192.168.10.10,fail,1558430858
2,192.168.10.10,fail,1558430864
2,192.168.10.10,fail,1558430878
 */
public class FlinkTM5_T_01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv);

        fsTableEnv.executeSql("create table dwd_order_info(" +
                "    id STRING," +
                "    user_id STRING," +
                "    order_status STRING," +
                "    create_time STRING, " +
                "    rowtime AS TO_TIMESTAMP(create_time)," +
                "    WATERMARK FOR rowtime AS rowtime" +
                "    ) with (" +
                "  'connector' = 'kafka',  " +
                "  'topic' = 'order_info'," +
                "  'properties.bootstrap.servers' = 'hadoop-single:9092', " +
                "  'properties.group.id' = 'flinktm5_t_01'," +
                "  'format' = 'json'," +
                "  'scan.startup.mode' = 'earliest-offset')"
        );

//        fsTableEnv.sqlQuery("select * from dwd_order_info").execute().print();

        Table resultT=fsTableEnv.sqlQuery("SELECT T.aid" +
                "            FROM dwd_order_info" +
                "            MATCH_RECOGNIZE (" +
                "              PARTITION BY user_id" +
                "              ORDER BY rowtime" +
                "              MEASURES" +
                "                A.id as aid,A.order_status as astatus," +
                "                B.id as bid,B.order_status as bstatus" +
                "              PATTERN (A{10}) WITHIN INTERVAL '5' MINUTE" +
                "              DEFINE" +
                "                A AS order_status = '1001'," +
                "                B AS order_status = '1005'" +
                "            ) AS T ");

        fsTableEnv.createTemporaryView("result_t",resultT);

        fsTableEnv.sqlQuery(" select aid from result_t").execute().print();

        //fsEnv.execute();

    }
}
